/*
 * Singularity Engine
 * Copyright (C) 2010 Vyacheslav Vorobyov, All rights reserved.
 * See LICENSE for full license information.
 */
package org.singularity.io;


import java.io.IOException;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

/**
 * @author vjache
 *
 */
public class Pump<T> implements Runnable {
	
	public static final Log LOG = LogFactory.getFactory().getInstance(Pump.class);

	private final CloseableIterator<T> _src;
	private final CloseablePipe<T> _dst;
	private final Object _id;
	
	private volatile boolean _cancel = false;
	private final Object _synch = new Object();

	/**
	 * 
	 */
	public Pump(
			CloseableIterator<T> aSource, 
			CloseablePipe<T> aDestin,
			Object aId) {
		_src = aSource;
		_dst = aDestin;
		_id = aId;
	}

	/* (non-Javadoc)
	 * @see java.lang.Runnable#run()
	 */
	@Override
	public void run() {
		synchronized(_synch)
		{
			Thread.currentThread().setName(toString());
			while(_src.hasNext())
			{
				try {
					final T next = _src.next();
					_dst.push(next);
					if(_cancel)
					{
						LOG.warn(String.format("Pump %s canceled.",this));
						break;
					}
				} catch (Exception e) {
					LOG.error(String.format("Error while push to destination %s.",_dst), e);
					break;
				}
			}
			
			try {
				_src.close();
			} catch (IOException e) {
				LOG.error(String.format("Error while closing source %s.",_src), e);
			}
			try {
				_dst.close();
			} catch (IOException e) {
				LOG.error(String.format("Error while closing destination %s.",_dst), e);
			}
		}
	}
	
	public void cancel()
	{
		_cancel = true;
		synchronized(_synch)
		{}
	}
	
	@Override
	public String toString() {
		return "Pump-"+_id;
	}
}
